package com.gnerv.sylvanas.module.sse.operation;

import com.gnerv.sylvanas.framework.common.model.sse.SseResult;
import com.gnerv.sylvanas.framework.core.tools.JsonTools;
import com.gnerv.sylvanas.module.sse.model.bo.SubscribeBO;
import com.gnerv.sylvanas.module.sse.storage.SseStorageService;
import lombok.AllArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;
import org.springframework.web.servlet.mvc.method.annotation.SseEmitter;

import java.io.IOException;
import java.util.List;

@Slf4j
@Service
@AllArgsConstructor
public class SseOperationService {

    private final SseStorageService sseStorageService;

    public SseEmitter createSseEmitter(String clientId) {
        SseEmitter sseEmitter = new SseEmitter(24 * 60 * 60 * 1000L);
        //完成后回调
        sseEmitter.onCompletion(() -> {
            log.info("[{}]发送完成...................", clientId);
        });
        //超时回调
        sseEmitter.onTimeout(() -> {
            log.info("[{}]连接超时...................", clientId);
        });
        //异常回调
        sseEmitter.onError(
                throwable -> {
                    try {
                        log.info("[{}]连接异常,{}", clientId, throwable.toString());
                        sseEmitter.send(SseEmitter.event()
                                .id(clientId)
                                .name("发生异常！")
                                .data("发生异常请重试！")
                                .reconnectTime(30000));
                    } catch (IOException e) {
                        log.info(e.getMessage());
                    }
                }
        );
        sseStorageService.insertSseEmitter(clientId, sseEmitter);
        return sseEmitter;
    }

    @Async(value = "ssePushExecutor")
    public void pushText(SseEmitter sseEmitter, SseResult<?> sseResult) {
        pushText(sseEmitter, JsonTools.toJsonString(sseResult));
    }

    @Async(value = "ssePushExecutor")
    public void pushText(SseEmitter sseEmitter, String text) {
        if (sseEmitter == null) {
            log.info("推送消息失败................... sseEmitter is null");
        }
        try {
            sseEmitter.send(text);
        } catch (Exception e) {
            log.info("推送消息失败...................");
        }
    }

    public boolean createSubscribe(SubscribeBO subscribeBO) {
        sseStorageService.insertSubscribe(subscribeBO.getClientId(), subscribeBO.getTopicList());
        return true;
    }

    public boolean updateSubscribe(SubscribeBO subscribeBO) {
        sseStorageService.deleteSseEmitter(subscribeBO.getClientId());
        sseStorageService.insertSubscribe(subscribeBO.getClientId(), subscribeBO.getTopicList());
        return true;
    }

    public boolean deleteSubscribe(SubscribeBO subscribeBO) {
        sseStorageService.deleteSubscribe(subscribeBO.getClientId(), subscribeBO.getTopicList());
        return true;
    }

    public SseEmitter selectSseEmitter(String clientId) {
        return sseStorageService.selectSseEmitter(clientId);
    }

    public List<SseEmitter> selectListSseEmitterByTopic(String topic) {
        return sseStorageService.selectListSseEmitterByTopic(topic);
    }

    @Async(value = "ssePushExecutor")
    public void heartbeat() {
        sseStorageService.getSseEmitterMap().forEach((key, value) -> {
            pushText(value, SseResult.build("heartbeat", true));
        });
    }

    public SseEmitter selectSseEmitterByClientIdAndTopic(String clientId, String topic) {
        return sseStorageService.selectSseEmitterByClientIdAndTopic(clientId, topic);
    }
}
